package com.sjc.lesson01;

import com.sjc.lesson01.domain.WordAndCount;
import com.sjc.lesson01.function.SplitWord;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class WordCount {
    public static void main(String[] args) throws Exception {
        // 1. 获取执行环境
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
        env.setParallelism(2);

        // 2. 获取数据源
        // 使用 flink提供的工具类，获取传递的参数
        ParameterTool parameterTool = ParameterTool.fromArgs(args);
        String hostname = parameterTool.get("hostname");
        int port = parameterTool.getInt("port");
        DataStreamSource<String> dataStream = env.socketTextStream(hostname, port);

        // 3. 数据处理
        SingleOutputStreamOperator<WordAndCount> wordCount = dataStream
                        .flatMap(new SplitWord())
                        .keyBy("word")
                        .sum("count");

        // 4. 数据输出
        wordCount.print().setParallelism(1);
        // 5. 启动任务
        env.execute("word count ....");
    }
}
